package com.byx.imserverutil.netty;

import com.byx.imserverutil.entity.MessageType;
import com.byx.imserverutil.entity.NettyChannel;
import com.byx.imserverutil.listener.OnMessageReceivedListener;
import com.byx.imserverutil.protobuf.ChatMessageProtobuf;
import com.byx.imserverutil.service.ImService;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.Inet4Address;
import java.util.List;

/**
 * NettyWebSocketImServiceImpl
 *
 * @author Superb
 * @date 2021/2/9 13:13
 */
public class NettyWebSocketImServiceImpl implements ImService {
    private static final Logger logger=LoggerFactory.getLogger(NettyWebSocketImServiceImpl.class);
    private int port;
    private OnMessageReceivedListener onMessageReceivedListener;
    private ServerBootstrap serverBootstrap;
    private Channel channel;
    private GroupInterface groupInterface;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workGroup;
    private volatile boolean isClosed;
    private boolean initialized;

    private static final NettyWebSocketImServiceImpl INSTANCE=new NettyWebSocketImServiceImpl();

    private NettyWebSocketImServiceImpl(){}

    public static NettyWebSocketImServiceImpl getInstance(){
        return INSTANCE;
    }
    /**
     * 初始化
     *
     * @param port                      端口号
     * @param onMessageReceivedListener 消息接收监听器
     * @return boolean
     */
    @Override
    public boolean initialize(int port, OnMessageReceivedListener onMessageReceivedListener) {
        this.port=port;
        this.onMessageReceivedListener=onMessageReceivedListener;
        initialized=true;
        return true;
    }

    /**
     * 启动服务
     */
    @Override
    public void start() {
        if(!initialized){
            logger.info("启动失败");
            return;
        }

        try{
            initServerBootstrap();
            ChannelFuture future = serverBootstrap.bind(port).sync();
            channel = future.channel();
            if (channel != null && channel.isOpen() && channel.isActive() && channel.isRegistered() && channel.isWritable()) {
                logger.debug(String.format("NettyWebSocketIMS启动成功，address = ws://%1$s:%2$d/websocket", Inet4Address.getLocalHost().getHostAddress(), port));
            } else {
                logger.debug("NettyWebSocketIMS启动失败");
            }

            future.awaitUninterruptibly();
            channel.closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 发送消息
     *
     * @param msg 消息实体
     */
    @Override
    public void sendMessage(ChatMessageProtobuf.Msg msg) {
        if (msg.getMessageType() == MessageType.SINGLE_CHAT_MSG){
            String toId= msg.getToId();
            NettyChannel nettyChannel=NettyChannelContainer.getInstance().getActiveChannel(toId);
//            System.out.println(nettyChannel);
            if(nettyChannel != null){
                //在线直接转发
                logger.info("正在转发消息");
                nettyChannel.getChannel().writeAndFlush(msg);
            }else {
                //对方不在线
                logger.info("对方不在线，转发失败");
            }
        }else if (msg.getMessageType() == MessageType.GROUP_CHAT_MSG){
            if (groupInterface != null){
                String toId = msg.getToId();
                List<String> members = groupInterface.queryMembersInGroup(Integer.parseInt(toId));
                for (String member : members){
                    if (!member.equals(msg.getFromId())){
                        NettyChannel nettyChannel=NettyChannelContainer.getInstance().getActiveChannel(member);
                        if (nettyChannel != null){
                            //在线直接转发
                            logger.info("正在转发消息");
                            nettyChannel.getChannel().writeAndFlush(msg);
                        }
                    }
                }
            }
        }

    }

    /**
     * 关闭服务，释放资源
     */
    @Override
    public void release() {
        closeChannel();
        closeServerBootstrap();
        isClosed=true;
    }

    public void setGroupInterface(GroupInterface groupInterface) {
        this.groupInterface = groupInterface;
    }

    private void initServerBootstrap(){
        try {
            if(null != serverBootstrap){
                closeServerBootstrap();// 先关闭之前的bootstrap
            }

            // boss线程池用于处理TCP连接，通常服务端开启的都是一个端口，所以线程数指定为1即可
            bossGroup = new NioEventLoopGroup(1);
            // work线程用于处理IO事件，需要多线程处理，不指定线程数，默认就是CPU核心数*2
            workGroup = new NioEventLoopGroup();
            serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    // 设置TCP接收缓冲区大小（字节数）
                    .option(ChannelOption.SO_RCVBUF, 32 * 1024)
                    // 服务端接受连接的队列长度，如果队列已满，客户端连接将被拒绝。默认值，Windows为200，其他为128
                    .option(ChannelOption.SO_BACKLOG, 256)
                    // 设置该选项以后，如果在两小时内没有数据的通信时，TCP会自动发送一个活动探测数据报文
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    // 设置禁用nagle算法，如果要求高实时性，有数据发送时就马上发送，就将该选项设置为true关闭Nagle算法；如果要减少发送次数减少网络交互，就设置为false等累积一定大小后再发送。默认为false
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new NettyWebSocketChannelInitializer(this));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void closeServerBootstrap(){
        try {
            if (serverBootstrap != null) {
                serverBootstrap.config().group().shutdownGracefully();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            serverBootstrap = null;
        }
    }

    private void closeChannel(){
        try {
            if (null != channel) {
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                try {
                    channel.eventLoop().shutdownGracefully();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            channel = null;
        }
    }
}
